package com.mjf.transformation

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object MapDemo {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val source: DataStream[String] = env.fromCollection(List(
      "a", "b", "c", "d", "e",
      "1", "2", "3", "4", "5"
    ))

    source.print("source")

    // 方式一：通过传入函数使用 map
//    val result1: DataStream[String] = source.map(line => {"map--->" + line})
    // 简写形式 1.函数体只有一行时可以省略花括号 2.如果在匿名函数中，参数只出现了一次，那么参数可以省略，在函数体使用参数的时候，用下划线代替
    val result1: DataStream[String] = source.map("map--->" + _)

    // 方式二：通过传入自定义函数类使用 map
    val result2: DataStream[(String, Long)] = source.map(new MyMapFunction)

    result1.print("函数")
    result2.print("自定义函数类")

    env.execute(MapDemo.getClass.getName)
    
  }
}

// 自定义函数类，继承 MapFunction
class MyMapFunction extends MapFunction[String, (String, Long)] {
  override def map(value: String): (String, Long) = (value, 1L)
}